/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.wsinfra.wsrm.impl;

import java.util.Date;
import java.util.Vector;

import javax.xml.soap.SOAPMessage;

import org.apache.xmlbeans.XmlCursor;

import cgl.narada.wsinfra.WsMessageFlow;
import cgl.narada.wsinfra.exception.MessageFlowException;
import cgl.narada.wsinfra.exception.ProcessingException;
import cgl.narada.wsinfra.schemas.addressing.ActionDocument;
import cgl.narada.wsinfra.schemas.addressing.EndpointReferenceType;
import cgl.narada.wsinfra.schemas.addressing.FromDocument;
import cgl.narada.wsinfra.schemas.addressing.RelatesToDocument;
import cgl.narada.wsinfra.schemas.soap.EnvelopeDocument;
import cgl.narada.wsinfra.schemas.soap.HeaderType;
import cgl.narada.wsinfra.schemas.wsrm.AckRequestedDocument;
import cgl.narada.wsinfra.schemas.wsrm.SequenceAcknowledgementDocument;
import cgl.narada.wsinfra.util.QNameLocator;
import cgl.narada.wsinfra.util.SoapEnvelopeConversion;
import cgl.narada.wsinfra.wsa.WsaEnvelopeCreator;
import cgl.narada.wsinfra.wsa.WsaProcessingFactory;
import cgl.narada.wsinfra.wsrm.WsrmAckOperations;
import cgl.narada.wsinfra.wsrm.WsrmElementAddition;
import cgl.narada.wsinfra.wsrm.WsrmElementCreation;
import cgl.narada.wsinfra.wsrm.WsrmNodeUtils;
import cgl.narada.wsinfra.wsrm.WsrmProcessingFactory;
import cgl.narada.wsinfra.wsrm.WsrmQNames;
import cgl.narada.wsinfra.wsrm.WsrmSequenceMonitor;
import cgl.narada.wsinfra.wsrm.WsrmSequencePolicies;
import cgl.narada.wsinfra.wsrm.storage.WsrmAuditStorageOperations;
import cgl.narada.wsinfra.wsrm.storage.WsrmPolicyStorageOperations;
import cgl.narada.wsinfra.wsrm.storage.WsrmProtocolStorageOperations;
import cgl.narada.wsinfra.wsrm.storage.WsrmSequenceInfo;
import cgl.narada.wsinfra.wsrm.storage.WsrmSequenceInfoStorageOperations;
import cgl.narada.wsinfra.wsrm.storage.WsrmStorageEventsFactory;
import cgl.narada.wsinfra.wsrm.storage.WsrmStorageException;
import cgl.narada.wsinfra.wsrm.storage.WsrmStorageService;
import cgl.narada.wsinfra.wsrm.storage.WsrmStorageWidget;

/** This processor processes Sequences and determines if 
    (a) Acknowledgements/Retransmissions need to be issued.
    (b) See if the Inactivity timeout on a sequence has expired, if so
    proceed to terminate the sequence. 
    
    @author Shrideep Pallickara and Sangmi Lee
    $Date: 2005/07/29 22:41:30 $
    $Revision: 1.3 $
*/


public class WsrmSequenceMonitorImpl extends Thread 
  implements WsrmSequenceMonitor {
  private SoapEnvelopeConversion soapEnvelopeConversion;
  private WsrmAckOperations wsrmAckOperations;
  private WsrmElementAddition wsrmElementAddition;
  private WsrmElementCreation wsrmElementCreation;
  
  private WsrmStorageService wsrmStorageService;
  private WsrmProtocolStorageOperations wsrmProtocolOps;
  private WsrmSequenceInfoStorageOperations wsrmSequenceInfoOps;
  private WsrmPolicyStorageOperations wsrmPolicyOps;
  private WsrmAuditStorageOperations wsrmAuditOps;
  private WsrmStorageEventsFactory storageEventsFactory;
  private WsrmNodeUtils wsrmNodeUtils;
  private WsrmQNames wsrmQNames;
  private QNameLocator qNameLocator;

  private WsaEnvelopeCreator wsaEnvelopeCreator;
  private boolean debug = true;
  
  private WsMessageFlow wsMessageFlow;
  private boolean keepRunning = true;
  private long SLEEP_INTERVAL = 15000;
  private String moduleName = "WsrmSequenceMonitor: ";
  
  protected WsrmSequenceMonitorImpl(String configInfo, 
				    WsMessageFlow wsMessageFlow) 
    throws WsrmStorageException {
    this.wsMessageFlow = wsMessageFlow;
    System.out.println(moduleName + "Using configuration file [" + configInfo 
		       + "]");
    wsrmStorageService = WsrmStorageService.getInstance(configInfo);
    wsrmProtocolOps = wsrmStorageService.getWsrmProtocolStorageOperations();
    wsrmSequenceInfoOps = 
      wsrmStorageService.getWsrmSequenceInfoStorageOperations(); 
    wsrmPolicyOps = wsrmStorageService.getWsrmPolicyStorageOperations();
    wsrmAuditOps =  wsrmStorageService.getWsrmAuditStorageOperations();
    
    storageEventsFactory = WsrmStorageEventsFactory.getInstance();
    
    wsrmAckOperations = WsrmProcessingFactory.getWsrmAckOperations();
    wsrmElementAddition = WsrmProcessingFactory.getWsrmElementAddition();
    wsrmElementCreation = WsrmProcessingFactory.getWsrmElementCreation();
    wsrmNodeUtils = WsrmProcessingFactory.getWsrmNodeUtils();

    wsaEnvelopeCreator = WsaProcessingFactory.getWsaEnvelopeCreator();
    soapEnvelopeConversion = SoapEnvelopeConversion.getInstance();
    //start();
    wsrmQNames = WsrmQNames.getInstance();

    qNameLocator = QNameLocator.getInstance();
    System.out.println(moduleName + "Initializations complete ...");
    
  }

  /** Begin services related to sequence monitoring viz. issue acknowledgements
      and initiate retransmissions for sequences. */
  public void startServices() {
    keepRunning = true;
    start();
  }
  
  /** Stop services related to sequence monitoring. */
  public void stopServices() {
    keepRunning = false;
  }

  public void run() {  
    while(keepRunning) {
      try {
	sleep(SLEEP_INTERVAL);
	cycleThroughSequences();
      } catch (MessageFlowException mfe) {
	System.out.println(moduleName + "Problems processing sequences. " + 
			   "MessageFlow problems " + mfe);
	mfe.printStackTrace();
	continue;
      } catch (WsrmStorageException wste) {
	System.out.println(moduleName + "Problems processing sequences " 
			   + wste);
	wste.printStackTrace();
	printTerminationInfo();
	break;
      } catch (InterruptedException e) {
	System.out.println(moduleName + "Problems sleeping " + e);
	printTerminationInfo();
	break;
      }
    }/* end while(keepLooping) */
    
    System.out.println(moduleName + "The thread is shutting down.");
  }

  private void printTerminationInfo() {
    System.out.println(moduleName + "Because of an error, Processing of this "
		       + "thread has been terrminated! No further " + 
		       "retransmissions/acks will be issued.");
  }
  
  private void cycleThroughSequences() 
    throws MessageFlowException, WsrmStorageException {
    String[] sentSequences = 
      wsrmSequenceInfoOps.getListOfActiveSequences(true);
    System.out.println(moduleName + "Processing sent sequences ");
    if (sentSequences == null) {
      if (debug) {
	System.out.println(moduleName + "No active Sent sequences to process");
      }
    }

    if (sentSequences != null) {
      for (int i= 0; i < sentSequences.length; i++) {
	String _sequenceId = sentSequences[i];
	WsrmSequenceInfo _wsrmSequenceInfo = 
	  wsrmSequenceInfoOps.getSequenceInfoUsingSequenceIdentifier(_sequenceId);

	boolean timedOut = checkInactivityTimeout(_wsrmSequenceInfo);
	boolean messagesAvailable=checkIfMessagesAvailable(_wsrmSequenceInfo);
	
	if (!timedOut && messagesAvailable) {
	  checkToIssueRetransmissions(_wsrmSequenceInfo);
	}
      }
    } 

    
    
    System.out.println(moduleName + "Processing received sequences ");
    String[] receivedSequences = 
      wsrmSequenceInfoOps.getListOfActiveSequences(false);
    if (receivedSequences == null) {
      if (debug) {
	System.out.println(moduleName + 
			   "No active received Sequences to process");
      }
    }

    if (receivedSequences != null) {
      for (int i= 0; i < receivedSequences.length; i++) {
	String _sequenceId = receivedSequences[i];
	WsrmSequenceInfo _wsrmSequenceInfo = 
	  wsrmSequenceInfoOps.getSequenceInfoUsingSequenceIdentifier(_sequenceId);
	boolean timedOut = checkInactivityTimeout(_wsrmSequenceInfo);
	boolean messagesAvailable=checkIfMessagesAvailable(_wsrmSequenceInfo);
	
	if (!timedOut && messagesAvailable) {
	  checkToIssueAcknowledgements(_wsrmSequenceInfo);
	}
      }
    } 
    
    
    
    
  }
  
  /** Checks to see if the inactivity interval has expired on the
      sequence in question. */
  public boolean checkInactivityTimeout(WsrmSequenceInfo wsrmSequenceInfo) 
    throws WsrmStorageException {
    boolean isTimedOut = false;
    String sequenceId = wsrmSequenceInfo.getSequenceIdentifier();
    
    WsrmSequencePolicies wsrmSequencePolicies = 
      wsrmSequenceInfo.getWsrmSequencePolicies();
    
    long inactivityTimeout = wsrmSequencePolicies.getInactivityTimeout().getInactivityTimeout().getMilliseconds().longValue();
    
    long timeOfLastActivity = wsrmSequenceInfo.getTimeOfLastActivity();
    long timeNow = System.currentTimeMillis();
    
    if (timeNow - timeOfLastActivity > inactivityTimeout) {
      String reason = "The inactivity timeout on the sequence has been " + 
	"exceeded. The last activity on this sequence was on " + 
	new Date(timeOfLastActivity) + ". The termination time is " + 
	new Date(timeNow);
      storageEventsFactory.timeoutSequence(wsrmSequenceInfo);
      storageEventsFactory.terminateSequence(wsrmSequenceInfo, reason);
      wsrmSequenceInfoOps.storeCreatedSequence(wsrmSequenceInfo);
      isTimedOut = true;
    } 
    
    return isTimedOut;
  }
  
  
  /** Checks to see if any messages have been stored for this sequence in
      question. */
  public boolean checkIfMessagesAvailable(WsrmSequenceInfo wsrmSequenceInfo)
    throws WsrmStorageException {
    String sequenceIdentifier = wsrmSequenceInfo.getSequenceIdentifier();

    boolean hasStoredElements = 
      wsrmProtocolOps.hasStoredElements(sequenceIdentifier);

    return hasStoredElements;
  }
  

  /** This method checks to see if acknowledgements should be issued on
      a specific sequence. If there is a neeed to do so, it proceeds to 
      issue acknowledgements on the sequence. */
  public void checkToIssueAcknowledgements(WsrmSequenceInfo wsrmSequenceInfo)
    throws MessageFlowException, WsrmStorageException {
    String sequenceIdentifier = wsrmSequenceInfo.getSequenceIdentifier();
    
    boolean ackRequested = false;
    long[] ackedMessageNumbers = 
      wsrmProtocolOps.getAcknowledgedMessageNumbers(sequenceIdentifier);
    long[] unackedMessageNumbers = 
      wsrmProtocolOps.getUnacknowledgedMessageNumbers(sequenceIdentifier, 
						      ackRequested);
    
    long[] acknowledgements = getCombinedArrayOfLong(ackedMessageNumbers,
						     unackedMessageNumbers);
    
    printAcknowledgementsInfo(sequenceIdentifier, unackedMessageNumbers, 
			      acknowledgements);

    /** Get the message numbers that need to be acknowledged. We also need to
	a method so that it retrieves the set of message numbers for negative
	acknowledgement */
    boolean positiveAcks = true;
    SequenceAcknowledgementDocument sequenceAckDocument =
      wsrmElementCreation.newSequenceAcknowledgement(sequenceIdentifier, 
						     acknowledgements, 
						     positiveAcks); 
    
        
    /** Process the acknowledgements */
    if (unackedMessageNumbers != null) {
      /** Proceed to issue sequence acknowledgement */
      issueAcknowledgement(wsrmSequenceInfo, sequenceAckDocument);
      wsrmProtocolOps.processAcknowledgementsOnSequence(sequenceIdentifier,
							unackedMessageNumbers);
    }    	
  }


  /** Print diagnostic information regarding the unackedMessagedNumbers and
      the set of messages that will be acknowledged */
  private void 
  printAcknowledgementsInfo(String sequenceId, long[] unackedMessageNumbers, 
			    long[] acknowledgements) {
    if (unackedMessageNumbers != null) {
      System.out.print(moduleName + "Unacked Message Numbers = ");
      for (int i=0; i < unackedMessageNumbers.length; i++) {
	System.out.print("[" + unackedMessageNumbers[i] + "]");
      }
      System.out.print("\n");
    }
    
    if (unackedMessageNumbers == null) {
      System.out.println(moduleName + "All message numbers on sequence [" +
			 sequenceId + 
			 "] have been ACKed. (Based on ackInterval).");
      return;
    } 
    
    if (acknowledgements == null) {
      System.out.println(moduleName + "No message numbers on sequence [" +
			 sequenceId + "] need to be ACKed");
      return;
    }
  }
  
  /** This method creates an enevelope based on the specified wsrmSequenceInfo
      and the SequenceAcknowledgement Document. */
  private void 
  issueAcknowledgement(WsrmSequenceInfo wsrmSequenceInfo,		       
		       SequenceAcknowledgementDocument sequenceAckDocument) 
    throws MessageFlowException {
    
    EndpointReferenceType sourceEpr = wsrmSequenceInfo.getSource();
    EndpointReferenceType sinkEpr = wsrmSequenceInfo.getDestination();
    FromDocument from = FromDocument.Factory.newInstance();
    from.setFrom(sinkEpr);
    
    ActionDocument action = null;
    
    String relatesToId = wsrmSequenceInfo.getAddressingIdentifier();
    RelatesToDocument relatesTo = RelatesToDocument.Factory.newInstance();
    relatesTo.addNewRelatesTo().setStringValue(relatesToId);
    
    EnvelopeDocument acknowledgementEnvelope = 
      wsaEnvelopeCreator.createSoapEnvelope(sourceEpr, from, 
					    action, relatesTo);
    
    wsrmElementAddition.addSequenceAcknowledgement(acknowledgementEnvelope,
						   sequenceAckDocument);
    
    enrouteToNetwork(acknowledgementEnvelope);
  }
  
  
  
  /** This method checks to see if retransmissions need to be issued on the
      specified sequence. If a need arises retransmissions are issued and
      the retransmission interval is reset. There should also be a way to
      ensure exponential backoff. */
  public void checkToIssueRetransmissions(WsrmSequenceInfo wsrmSequenceInfo) 
    throws WsrmStorageException, MessageFlowException  {
    String sequenceId = wsrmSequenceInfo.getSequenceIdentifier();
    long[] messageNumbersNotAcked =
      wsrmProtocolOps.getUnacknowledgedMessageNumbers(sequenceId, false);
    
    if (messageNumbersNotAcked == null) {
      System.out.println(moduleName + "No message numbers on sequence [" +
			 sequenceId + "] need to be Retransmitted");
      return;
    }
    
    WsrmStorageWidget[] widgetsToRetransmit = 
      wsrmProtocolOps.getStoredElements(sequenceId, messageNumbersNotAcked);
    
    if (widgetsToRetransmit == null) {
      System.out.println(moduleName + "Widgets obtained for retransmissions " +
			 "are NULL. This should NOT have been the case under" +
			 "ANY circumstance");
      return;
    }
    
    int numOfWidgetsToRetransmit = widgetsToRetransmit.length;
    AckRequestedDocument ackRequestedDocument;
    /** Add AckRequested element to the envelope document */
    if (wsrmSequenceInfo.hasLastMessageInfo() ) {
      long lastMessageNumber = wsrmSequenceInfo.getLastMessageNumber();
      ackRequestedDocument = 
	wsrmElementCreation.newAckRequested(sequenceId, lastMessageNumber);
    } else {
      ackRequestedDocument = 
	wsrmElementCreation.newAckRequested(sequenceId);
    }

    for (int i=0; i < numOfWidgetsToRetransmit; i++) {
      retransmitMessage(widgetsToRetransmit[i], ackRequestedDocument); 
      long resetInterval = computeNewRetransmissionInterval(wsrmSequenceInfo);
      wsrmProtocolOps.resetAckInterval(widgetsToRetransmit[i], resetInterval);
    }
  }
  
  private void retransmitMessage(WsrmStorageWidget wsrmStorageWidget,
				 AckRequestedDocument ackRequestedDocument) 
    throws MessageFlowException {
    System.out.println(moduleName + "Retransmissing MessageNumber=" +
		       wsrmStorageWidget.getMessageNumber() );
    

    EnvelopeDocument retransmitDocument = wsrmStorageWidget.getEnvelope();

    boolean hasAckRequested = hasAckRequestedElement(retransmitDocument);
    if (!hasAckRequested) {
      wsrmElementAddition.addAckRequested(retransmitDocument, 
					  ackRequestedDocument);
    }

    /** Check to see if the document contains an AckRequested element. If it 
	does do nothing, otherwise add an AckRequested element to the
	document. */
    enrouteToNetwork(retransmitDocument);
  }
  
  

  private void enrouteToNetwork(EnvelopeDocument envelopeDocument)
    throws MessageFlowException { 
    
    SoapEnvelopeConversion soapEnvelopeConversion = 
      SoapEnvelopeConversion.getInstance();
    SOAPMessage soapMessage = null;
    try {
      soapMessage =  soapEnvelopeConversion.getSOAPMessage(envelopeDocument);
    } catch (ProcessingException ex) {
      String errorReport ="Problems converting to javax.xml.SOAPMessage\n"
	+ ex.toString();
      throw new MessageFlowException(errorReport);
    }
    wsMessageFlow.enrouteToNetwork(soapMessage);
  }


  

  
  /** Computes a new retransmission interval based on a sequence's policies. */
  private long 
  computeNewRetransmissionInterval(WsrmSequenceInfo wsrmSequenceInfo) {
    WsrmSequencePolicies wsrmSequencePolicies = 
      wsrmSequenceInfo.getWsrmSequencePolicies();
    
    long retransmissionInterval = wsrmSequencePolicies.getRetransmissionInterval().getBaseRetransmissionInterval().getMilliseconds().longValue();

    retransmissionInterval += System.currentTimeMillis();

    return retransmissionInterval;
  }

  private long retrieveAckInterval(WsrmSequenceInfo wsrmSequenceInfo) {
    long ackInterval = 40000; /** initialize this to a default */
    
    return ackInterval;
  }


  /** Combines the two arrays to create a unique array without any
      duplicate values in the combined array. This will also eliminate any
      duplicates that might exist in the original arrays */
  private long[]
  getCombinedArrayOfLong(long[] first, long[] second) {
    if (first == null) {
      return second;
    }
    
    if (second == null) {
      return first;
    }
    
    Vector combinedVector = new Vector();
    for (int i=0; i < first.length; i++) {
      addToVector(combinedVector, first[i]);
    }
    
    for (int i=0; i < second.length; i++) {
      addToVector(combinedVector, second[i]);
    }
    
    if (combinedVector.size() == 0) {
      return null;
    }
    
    int numOfElements = combinedVector.size();
    long[] mergedArray = new long[numOfElements];
    for (int i=0; i < numOfElements; i++) {
      mergedArray[i] = ((Long)combinedVector.elementAt(i)).longValue();
    }
    
    return mergedArray;
  }
  
  
  private void addToVector(Vector vector, long value) {
    Long toAdd = new Long(value);
    if (vector.contains(toAdd)) {
      return;
    }
    vector.addElement(toAdd);
  }

  /** Checks to see if the AckRequested element is present in the
      envelope. */
  private boolean hasAckRequestedElement(EnvelopeDocument envelopeDocument) {
    HeaderType header = envelopeDocument.getEnvelope().getHeader();
    XmlCursor headerCursor = header.newCursor();
    
    boolean elementFound = false;
     /** Locate the ackRequested element */
    elementFound = 
      qNameLocator.locateQName(headerCursor, wsrmQNames.getAckRequested());

    headerCursor.dispose();
    return elementFound;
  }
}
